package com.cttic.kafka2hbase.config;


import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.client.Admin;
import org.apache.hadoop.hbase.client.Connection;
import org.apache.hadoop.hbase.client.ConnectionFactory;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.stereotype.Component;

import java.io.IOException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

@Configuration
public class HBaseConfig {

    @Value("${hbase.zookeeper.quorum}")
    private String hbaseZkQuorum;

    @Bean
    public Connection connectionHBase(){
        org.apache.hadoop.conf.Configuration conf = HBaseConfiguration.create();

        ExecutorService pool = Executors.newScheduledThreadPool(20);	//设置连接池
        try {
            conf.set("hbase.zookeeper.quorum",hbaseZkQuorum);
            conf.set("hbase.zookeeper.property.clientPort","2181");
            Connection Connection = ConnectionFactory.createConnection(conf, pool);
            return Connection;
        } catch (IOException e) {
            //logger.error("HbaseUtils实例初始化失败！错误信息为：" + e.getMessage(), e);
            e.printStackTrace();
            return null;
        }

    }


    @Bean
    public Admin admin(){
        try {
            return connectionHBase().getAdmin();
        } catch (IOException e) {
            e.printStackTrace();
            return null;
        }
    }

}
